package com.smile.sink;

//public static class MyJdbcSink extends RichSinkFunction<SensorReading> {
//    Connection conn = null;
//    PreparedStatement insertStmt = null;
//    PreparedStatement updateStmt = null;
//    // open 主要是创建连接
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test",
//                "root", "123456");
//        // 创建预编译器，有占位符，可传入参数
//        insertStmt = conn.prepareStatement("INSERT INTO sensor_temp (id, temp) VALUES
//        (?, ?)");
//        updateStmt = conn.prepareStatement("UPDATE sensor_temp SET temp = ? WHERE id
//                = ?");
//    }
//    // 调用连接，执行 sql
//    @Override
//    public void invoke(SensorReading value, Context context) throws Exception {
//        // 执行更新语句，注意不要留 super
//        updateStmt.setDouble(1, value.getTemperature());
//        updateStmt.setString(2, value.getId());
//        updateStmt.execute();
//        // 如果刚才 update 语句没有更新，那么插入
//        if (updateStmt.getUpdateCount() == 0) {
//            insertStmt.setString(1, value.getId());
//            insertStmt.setDouble(2, value.getTemperature());
//            insertStmt.execute();
//        }
//    }
//    @Override
//    public void close() throws Exception {
//        insertStmt.close();
//        updateStmt.close();
//        conn.close();
//    }
//}

//在 main 方法中增加，把明细保存到 mysql 中
//dataStream.addSink(new MyJdbcSink())

